package co.yixiang.modules.shop.rest;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class TDMQConsumerListener implements ServletContextListener {
	private final CommissionMainController CommissionController;
	private final CommissionNewController CommissionNewController;
	 @Override
	    public void contextInitialized(ServletContextEvent sce) {
		 PulsarClient client =null;
			SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
			try {
				System.out.printf("进入TDMQconsumer队列===========");
				log.info("进入TDMQconsumer队列===========");
				//initTDMQ();
				
				
				 /* client = PulsarClient.builder().serviceUrl("pulsar://172.21.0.14:6000/")
				  .listenerName("custom:pulsar-oonjr2vzdo/vpc-r9z7v0vc/subnet-6j2pevlp")
				  .authentication(AuthenticationFactory.token(
				  "eyJrZXlJZCI6InB1bHNhci1vb25qcjJ2emRvIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9jdXN0b20ifQ.IKE26NFA1HbWE8C_xw3cZTgiNge1wwV-G1eXITOr_aM"
				  )) .build();*/





			  client = PulsarClient.builder().serviceUrl("http://pulsar-p3e9omdmeb.tdmq-jpnc.ap-gz.qcloud.tencenttdmq.com:5121")
			  //.listenerName("custom:pulsar-p3e9omdmeb/vpc-fs65a66n/subnet-jtagylkm")
			  .authentication(AuthenticationFactory.token(
			  "eyJrZXlJZCI6InB1bHNhci1wM2U5b21kbWViIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJwdWxzYXItcDNlOW9tZG1lYl9qdXhpX2N1c3RvbWVyIn0.MSaQthTgmv4R5yZbUS9KVUmy4nqirc9F638lDtknFL8"
			  )) .build();
				Consumer<byte[]> consumer = client.newConsumer()
						 //.topic("persistent://pulsar-oonjr2vzdo/orderCommission/orderCommission_tdmq")
						.topic("persistent://pulsar-p3e9omdmeb/orderCommission_juxi/orderCommission_juximsg")
						 .messageListener(new  MessageListener<byte[]>() {
						        @Override
						         public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
						                 //处理消息
						        	try {
							        	log.info("进入consumer队列发送信息开始时间=========" + df.format(new Date()));
							        	log.info("进入consumer队列发送信息msg=========" + msg.getData());
										if (new String(msg.getData()).indexOf("user_id") != -1) {
											//CommissionController.updateParentUserCount(new String(msg.getData()).split(":")[1]);
											CommissionNewController.updateParentUserCount(new String(msg.getData()).split(":")[1]);
										} else {
											//CommissionController.checkOrderGoodsclass(new String(msg.getData()));
											CommissionNewController.checkOrderGoodsclass(new String(msg.getData()));
										}
										System.out.println("进入队列发送信息结束时间=========" + df.format(new Date()));
										log.info("进入队列发送信息结束时间=========" + df.format(new Date()));
										consumer.acknowledge(msg);
						        	} catch (Exception e) {
									// 处理出错
									System.out.println("报错consumerMessage1111" + e.getMessage());
									log.error("报错consumerMessage1111" + e.getMessage());
						        	}
						        }
						   })
						  .subscriptionName("consumer003")
						  .subscribe();





			} catch (Exception e) {
				System.out.println("报错consumerMessage" + e.getMessage());
				log.error("报错consumerMessage" + e.getMessage());
				e.printStackTrace();
			}
	    }

	    @Override
	    public void contextDestroyed(ServletContextEvent sce) {
	        System.out.println("web容器关闭, context对象被销毁..."); 
	    }

	public static void main(String args[]){
		
	}

	//public static void main(String[] argv) throws Exception {
		/*
		 * try {
		 * 
		 * 
		 * PulsarClient client =
		 * PulsarClient.builder().serviceUrl("pulsar://172.16.0.9:6000/")
		 * .listenerName("custom:pulsar-bkqqdzvx3z/vpc-9lez2cx9/subnet-ozhd1fdo")
		 * .authentication(AuthenticationFactory.token(
		 * "eyJrZXlJZCI6InB1bHNhci1ia3FxZHp2eDN6IiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9jdXN0b20ifQ.ZDRhX2EnMWducHugHH2LWgAgd8DfZ8-xjWhQ3ojuEZU"
		 * )) .build(); String s ="111111"; Producer<byte[]> producer =
		 * client.newProducer()
		 * .topic("persistent://pulsar-bkqqdzvx3z/orderCommission/orderCommission_msg").
		 * create(); producer.send(s.getBytes());
		 * 
		 * Consumer consumer = client.newConsumer().topic(
		 * "persistent://pulsar-bkqqdzvx3z/orderCommission/orderCommission_msg") //
		 * .subscriptionType(SubscriptionType.Shared) //
		 * .enableRetry(true)默认关闭，如果需要重试则开启
		 * .subscriptionName("consumer001").subscribe(); while (true) { // 等待接收消息
		 * Message msg = consumer.receive(); try {
		 * System.out.printf("Message received: %s", new String(msg.getData())); //
		 * 消息ACK consumer.acknowledge(msg); } catch (Exception e) { // 处理出错 //
		 * enableRetry=true 才能使用重试方法 consumer.reconsumeLater(msg, 1000L,
		 * TimeUnit.MILLISECONDS);// 消息按照指定的延迟时间重试 //
		 * consumer.reconsumeLater(msg);按照延迟等级进行重试，多次重试默认自增延迟等级 //
		 * consumer.reconsumeLater(msg, 1);按照指定的延迟等级进行重试 // delayLevel =1 代表1s, //
		 * delayLevel =2 代表5s, //
		 * 默认延迟等级"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" }
		 * CommissionMain.checkOrderGoodsclass(new String(msg.getData())); } } catch
		 * (Exception e) { System.out.printf("报错", e.getMessage()); }
		 */
	//}

}
